package cloud.xiguapi.ubas.analysis.hotitems.model;

import cloud.xiguapi.ubas.model.HotItemsViewCount;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-17 上午 10:13
 * desc:
 */
@Slf4j
public class TopNHotItems extends KeyedProcessFunction<Tuple, HotItemsViewCount, TopNHotItemsResult> {

    /**
     * top n的大小
     */
    private final Integer topSize;
    @Resource(name = "jedis")
    private Jedis jedis;
    /**
     * 定义列表状态, 保存当前窗口内所有输出的ItemViewCount
     */
    private transient ListState<HotItemsViewCount> itemViewCountListState;

    public TopNHotItems() {
        this.topSize = 10;
    }

    public TopNHotItems(Integer topSize) {
        this.topSize = topSize;
    }

    @Override
    public void open(Configuration parameters) {
        itemViewCountListState = getRuntimeContext()
                .getListState(new ListStateDescriptor<>("item-view-count-list", HotItemsViewCount.class));
    }


    @Override
    public void processElement(HotItemsViewCount value, Context ctx, Collector<TopNHotItemsResult> out) throws Exception {
        // 每来一条数据, 存入list中, 并注册定时器
        itemViewCountListState.add(value);
        // Flink底层是根据时间戳来注册定时器的, 相同窗口的时间戳是一致的, 相同的窗口只会注册一个时间戳
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<TopNHotItemsResult> out) throws Exception {
        // 定时器触发, 已收集到所有数据, 排序输出
        ArrayList<HotItemsViewCount> hotItemsViewCounts = Lists.newArrayList(itemViewCountListState.get().iterator());

        // 自定义排序规则
        hotItemsViewCounts.sort((o1, o2) -> o2.getCount().compareTo(o1.getCount()));

        List<TopNHotItemsResult.TopNHotItemsDetails> details = new ArrayList<>();

        // 遍历列表, 取top n输出, 需要注意遍历次数, 可能itemViewCounts中数量没有topSize个会造成越界
        for (int i = 0; i < Math.min(topSize, hotItemsViewCounts.size()); i++) {
            HotItemsViewCount currentItemViewCount = hotItemsViewCounts.get(i);
            details.add(TopNHotItemsResult.TopNHotItemsDetails.builder()
                    .itemId(currentItemViewCount.getItemId())
                    .count(currentItemViewCount.getCount())
                    .build());
        }

        // 控制输出频率
        Thread.sleep(200L);

        TopNHotItemsResult result = TopNHotItemsResult.builder()
                .windowEnd(new Timestamp(timestamp - 1).toString())
                .details(details)
                .build();

        // 输出格式化后的结果
        out.collect(result);
    }
}
